#[cfg(feature = "postgres")]
use crate::cache::get_pg_pool;
use crate::cache::{get_mysql_pool, IndexChange};
use crate::keys::VddKey;
use crate::sqlx::SqlxDefineTrait;
use crate::{
    cache, delete_by_key_head_in_trash, field_db, log_db, move_to_trash, record_db, run_by_store,
    sqlx_run_by_store, CF_TRASH, ROCKSDB,
};
use rust_rocksdb::WriteBatchWithTransaction;
use std::collections::HashMap;
use std::sync::Arc;
use visit_dd_core::model::{DataStoreEngine, VddModel, VddModelRef, VddSchema};
use visit_dd_core::proj::{VddLogPart, APP_CONFIG};
use visit_dd_core::query::WebTreeNode;
use visit_dd_core::record::RecordMeta;
use visit_dd_core::{alog, Ares, Astr, FieldId, RecordId, UserId, EMPTY_RID};

pub async fn m_load(mid: u64) -> VddModelRef {
    run_by_store!(impl_rdb::m_load, impl_sqlx::m_load, mid)
}

pub async fn m_select_by_uid(uid: &UserId) -> Vec<VddModelRef> {
    run_by_store!(impl_rdb::m_select_by_uid, impl_sqlx::m_select_by_uid, uid)
}

pub async fn m_select_ids_by_uid(uid: &UserId) -> Vec<u64> {
    run_by_store!(
        impl_rdb::m_select_ids_by_uid,
        impl_sqlx::m_select_ids_by_uid,
        uid
    )
}

pub async fn m_save(model: &VddModel, is_update: bool) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::m_save(&mut batch, &mut index_changes, model)
    } else {
        let ds = APP_CONFIG.meta_store.clone().unwrap();
        match ds.store_engine {
            #[cfg(feature = "postgres")]
            DataStoreEngine::Postgres => {
                impl_sqlx::m_save(&get_pg_pool(ds.as_ref()).await, model, is_update).await?
            }
            DataStoreEngine::Mysql => {
                impl_sqlx::m_save(&get_mysql_pool(ds.as_ref()).await, model, is_update).await?
            }
        };
    };
    cache::update_label_cache(&mut batch, model.id, &model.name);
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(model.uid.to_owned(), "保存".into()),
            VddLogPart::LpModel(model.id),
        ),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
        cache::invalidate_schema(model.id).await;
    }
    ares
}

pub async fn m_delete(model: &VddModel, operator: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    let field_ids = field_db::f_select_field_ids(model.id).await;
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::m_delete(&mut batch, &mut index_changes, model, &field_ids)
    } else {
        let ds = APP_CONFIG.meta_store.clone().unwrap();
        match ds.store_engine {
            #[cfg(feature = "postgres")]
            DataStoreEngine::Postgres => {
                impl_sqlx::m_delete(&get_pg_pool(ds.as_ref()).await, model).await?
            }
            DataStoreEngine::Mysql => {
                impl_sqlx::m_delete(&get_mysql_pool(ds.as_ref()).await, model).await?
            }
        };
    };
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator.clone(), "删除".into()),
            VddLogPart::LpModel(model.id),
        ),
    );
    // 记录基本信息
    let key = VddKey::IndexModel2Record(model.id).to_string();
    cache::get_index(&key).iter().for_each(|x| {
        let key2 = VddKey::EntityRecordMeta(RecordId(*x));
        move_to_trash(&mut batch, &key2, None);
    });
    // 模型到记录的索引
    batch.delete(&key);
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
        cache::invalidate_schema(model.id).await;
    }
    ares
}

pub async fn m_count_fields(model_id: u64) -> usize {
    run_by_store!(
        impl_rdb::m_count_fields,
        impl_sqlx::m_count_fields,
        model_id
    )
}

pub async fn m_count_all() -> usize {
    run_by_store!(impl_rdb::m_count_all, impl_sqlx::m_count_all)
}

pub(crate) async fn m_schema(model_id: u64) -> Result<Arc<VddSchema>, Astr> {
    run_by_store!(impl_rdb::m_schema, impl_sqlx::m_schema, model_id)
}

pub async fn m_select_sharing(exclude_user: &UserId) -> Vec<VddModelRef> {
    run_by_store!(
        impl_rdb::m_select_sharing,
        impl_sqlx::m_select_sharing,
        exclude_user
    )
}

/// 保存模型和字段的标记
pub async fn m_save_tags(model: &VddModel, model_tag: &str, tag_map: &HashMap<Astr, u64>) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::m_save_tags(&mut batch, model, model_tag, tag_map)
    } else {
        let ds = APP_CONFIG.meta_store.clone().unwrap();
        let _ = match ds.store_engine {
            #[cfg(feature = "postgres")]
            DataStoreEngine::Postgres => {
                get_pg_pool(ds.as_ref())
                    .await
                    .m_save_tags(model, model_tag, tag_map)
                    .await
            }
            DataStoreEngine::Mysql => {
                get_mysql_pool(ds.as_ref())
                    .await
                    .m_save_tags(model, model_tag, tag_map)
                    .await
            }
        };
    };
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(model.uid.clone(), "更新标记".into()),
            VddLogPart::LpModel(model.id),
        ),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::invalidate_schema(model.id).await;
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn m_select_tags_to_map(
    with_model: bool,
    with_field: bool,
    model_id: Option<u64>,
) -> HashMap<u64, Astr> {
    run_by_store!(
        impl_rdb::m_select_tags_to_map,
        impl_sqlx::m_select_tags_to_map,
        with_model,
        with_field,
        model_id
    )
}

pub async fn m_is_all_field_have_tag(model_id: u64, model_tag: &str) -> bool {
    run_by_store!(
        impl_rdb::m_is_all_field_have_tag,
        impl_sqlx::m_is_all_field_have_tag,
        model_id,
        model_tag
    )
}

pub async fn m_clear_tags(model_id: u64) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::m_clear_tags(&mut batch, model_id)
    } else {
        let ds = APP_CONFIG.meta_store.clone().unwrap();
        let field_ids = field_db::f_select_field_ids(model_id).await;
        match ds.store_engine {
            #[cfg(feature = "postgres")]
            DataStoreEngine::Postgres => {
                get_pg_pool(ds.as_ref())
                    .await
                    .m_clear_tags(model_id, &field_ids)
                    .await?
            }
            DataStoreEngine::Mysql => {
                get_mysql_pool(ds.as_ref())
                    .await
                    .m_clear_tags(model_id, &field_ids)
                    .await?
            }
        };
    }
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::invalidate_schema(model_id).await;
    }
    ares
}

pub async fn m_load_by_tag(tag: &str) -> Option<VddModelRef> {
    run_by_store!(impl_rdb::m_load_by_tag, impl_sqlx::m_load_by_tag, tag)
}

pub async fn m_is_tag_unique(tag: &str, model_id: u64) -> bool {
    run_by_store!(
        impl_rdb::m_is_tag_unique,
        impl_sqlx::m_is_tag_unique,
        tag,
        model_id
    )
}

pub async fn m_drop_from_trash(model_id: u64) -> Ares {
    match APP_CONFIG.meta_store.clone() {
        None => impl_rdb::drop_from_trash(model_id),
        Some(..) => {
            let field_ids = sqlx_run_by_store!(impl_sqlx::m_drop_from_trash, model_id).unwrap();
            let mut batch = WriteBatchWithTransaction::<true>::default();
            let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
            // 字段值
            field_ids.iter().for_each(|field_id| {
                let key_head = VddKey::BoundFieldValue(field_id.clone(), EMPTY_RID);
                delete_by_key_head_in_trash(&mut batch, &key_head);
            });
            let mut iter = ROCKSDB.raw_iterator_cf(&cf_trash);
            // 值基本信息
            iter.seek_to_first();
            while iter.valid() {
                let (k, v) = iter.item().unwrap();
                let key = String::from_utf8(k.to_vec()).unwrap();
                let key = VddKey::try_from(key.as_str());
                if let Ok(key) = key {
                    match key {
                        VddKey::EntityRecordMeta(..) => {
                            let meta = serde_json::from_slice::<RecordMeta>(v).unwrap();
                            if meta.model_id == model_id {
                                batch.delete_cf(
                                    &cf_trash,
                                    &VddKey::EntityRecordMeta(meta.id).to_string(),
                                );
                            }
                        }
                        _ => {}
                    }
                }
                iter.next();
            }
            ROCKSDB.write(batch).map_err(|e| e.to_string().into())
        }
    }
}

pub async fn m_restore(
    model_id: u64,
    field_ids: &[FieldId],
    with_fv: bool,
    operator: UserId,
) -> Ares {
    let ares = match APP_CONFIG.meta_store.clone() {
        None => impl_rdb::restore(model_id, field_ids),
        Some(..) => sqlx_run_by_store!(impl_sqlx::m_restore, model_id, field_ids, operator),
    };
    if ares.is_ok() && with_fv {
        record_db::restore(model_id, field_ids, &Vec::<RecordId>::new()).await
    } else {
        ares
    }
}

/// 校验一个模型是否有定义项目在回收站中
pub async fn design_in_trash(model_id: u64) -> bool {
    run_by_store!(
        impl_rdb::design_in_trash,
        impl_sqlx::design_in_trash,
        model_id
    )
}

// 一个用户回收站里的数据树形结构
pub async fn tree_of_trash(uid: UserId) -> Vec<WebTreeNode> {
    let mut schemas = Vec::<Arc<VddSchema>>::new();
    for model_id in m_select_ids_by_uid(&uid).await.iter() {
        schemas.push(cache::get_schema(*model_id).await.unwrap());
    }
    let nodes = run_by_store!(
        impl_rdb::tree_of_trash,
        impl_sqlx::tree_of_trash,
        uid,
        &schemas
    );
    // let mut fv_count_map = HashMap::<u64, u64>::new();
    // let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
    // let mut iter = ROCKSDB.raw_iterator_cf(&cf_trash);
    // // 回收站中的字段值
    // iter.seek_to_first();
    // while iter.valid() {
    //     let (k, v) = iter.item().unwrap();
    //     let key = String::from_utf8(k.to_vec()).unwrap();
    //     // tracing::warn!("key:{}", key);
    //     match judge_key_type(&key) {
    //         Some(kp) => match kp {
    //             KpBound(bt) => match bt {
    //                 BoundType::BtFv(fid) => {
    //                     let mut count = *fv_count_map.get(&fid.0).unwrap_or(&0);
    //                     count += 1;
    //                     fv_count_map.insert(fid.0, count);
    //                 }
    //                 _ => {}
    //             },
    //             _ => {}
    //         },
    //         None => {}
    //     }
    //     iter.next();
    // }
    // let mut empty_model_id = HashSet::<u64>::new(); // 字段值在回收站中为空的模型id
    // for mut m_node in nodes.iter_mut() {
    //     let mut empty = true;
    //     if m_node.children.is_some() {
    //         let mut childs = m_node.children.as_mut().unwrap();
    //         for mut f_node in &mut childs.iter_mut() {
    //             let fv_count = fv_count_map.get(&f_node.key).unwrap_or(&0);
    //             // Arc::get_mut(&mut f_node).unwrap().code = fv_count.to_string().into();
    //             if empty {
    //                 empty = *fv_count == 0
    //             };
    //         }
    //     }
    //     if empty {
    //         empty_model_id.insert(m_node.key);
    //     }
    // }
    // nodes.retain(|node| !empty_model_id.contains(&node.key));
    nodes
}

mod impl_rdb {
    use crate::cache::IndexChange;
    use crate::keys::VddKey;
    use crate::{
        cache, delete_by_key_head, delete_by_value, field_db, load_by_key_head,
        load_by_key_head_with_key, move_to_trash, move_to_trash_by_key_head, CF_META, CF_TRASH,
        CF_UNIQUE, ROCKSDB,
    };
    use crate::{count_in_key_head, delete_by_key_head_in_trash};
    use rust_rocksdb::WriteBatchWithTransaction;
    use serde_json::json;
    use std::collections::HashMap;
    use std::sync::Arc;
    use visit_dd_core::field::VddField;
    use visit_dd_core::i18n_text::E_ROCKSDB;
    use visit_dd_core::model::{VddModel, VddModelRef, VddSchema};
    use visit_dd_core::query::WebTreeNode;
    use visit_dd_core::record::RecordMeta;
    use visit_dd_core::{
        extract_last_str_from_key, i18n, Ares, Astr, FieldId, UserId, AC_ADD, AC_DELETE, EMPTY_RID,
    };

    pub(super) fn m_load(mid: u64) -> VddModelRef {
        let k = VddKey::EntityModel(mid).to_string();
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let v = ROCKSDB.get_cf(cf_meta, k).unwrap().unwrap();
        let model = serde_json::from_slice::<VddModel>(&v).unwrap();
        Arc::new(model)
    }

    pub(super) fn m_select_by_uid(uid: &UserId) -> Vec<VddModelRef> {
        let mut model_vec = Vec::<VddModelRef>::new();
        let index_key = VddKey::IndexUser2Model(uid.clone()).to_string();
        let m_index = cache::get_index(&index_key);
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let keys = m_index
            .iter()
            .map(|m| (cf_meta, VddKey::EntityModel(*m).to_string()));
        ROCKSDB.multi_get_cf(keys).iter().for_each(|item| {
            if let Ok(Some(vu8)) = item {
                let model = serde_json::from_slice::<VddModel>(vu8).unwrap();
                model_vec.push(model.into());
            }
        });
        model_vec
    }

    pub(super) fn m_select_ids_by_uid(uid: &UserId) -> Vec<u64> {
        let index_key = VddKey::IndexUser2Model(uid.clone()).to_string();
        cache::get_index(&index_key)
    }

    pub(super) fn m_save(
        batch: &mut WriteBatchWithTransaction<true>,
        index_changes: &mut Vec<IndexChange>,
        model: &VddModel,
    ) {
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        batch.put_cf(
            cf_meta,
            VddKey::EntityModel(model.id).to_string(),
            json!(model).to_string(),
        );
        cache::update_label_cache(batch, model.id, &model.name);
        index_changes.push(IndexChange::new(
            VddKey::IndexUser2Model(model.uid.clone()),
            AC_ADD,
            model.id,
        ));
    }

    pub(super) fn m_delete(
        batch: &mut WriteBatchWithTransaction<true>,
        index_changes: &mut Vec<IndexChange>,
        model: &VddModel,
        field_ids: &[FieldId],
    ) {
        //删除字段定义

        field_ids.iter().for_each(|x| {
            let key = VddKey::EntityField(FieldId(x.0));
            move_to_trash_by_key_head(batch, &key);
            //删除字段的标记
            delete_by_value(
                batch,
                &VddKey::UniqueFieldTag("".into(), "".into()),
                &x.0.to_le_bytes(),
            );
        });
        //用户模型索引中的关系
        index_changes.push(IndexChange::new(
            VddKey::IndexUser2Model(model.uid.clone()),
            AC_DELETE,
            model.id,
        ));
        // 分享规则
        delete_by_key_head(batch, &VddKey::BoundVisitGroup(model.id, UserId(0)));
        // 访问请求
        delete_by_key_head(batch, &VddKey::BoundVisitRequest(model.id, UserId(0)));
        // 标记
        let key = VddKey::UniqueModelTag("".into());

        delete_by_value(batch, &key, &model.id.to_le_bytes());
        //删除该模型到字段的索引
        batch.delete(VddKey::IndexModel2Field(model.id).to_string());
        // 模型日志
        batch.delete(VddKey::IndexModel2ActionLog(model.id).to_string());
        // 模型
        let key = VddKey::EntityModel(model.id);
        move_to_trash(batch, &key, None);
    }

    pub(super) fn m_count_fields(model_id: u64) -> usize {
        let key = VddKey::IndexModel2Field(model_id).to_string();
        let index = cache::get_index(&key);
        index.len()
    }

    pub(super) fn m_count_all() -> usize {
        count_in_key_head(&VddKey::EntityModel(0), |_| true)
    }

    pub(super) fn m_schema(model_id: u64) -> Result<Arc<VddSchema>, Astr> {
        let model = m_load(model_id);
        let fields = field_db::impl_rdb::f_select_by_model(model_id);
        let tags = m_select_tags_to_map(true, true, Some(model_id));
        Ok(Arc::new(VddSchema {
            model,
            fields,
            tags,
        }))
    }

    pub(super) fn m_select_sharing(exclude_user: &UserId) -> Vec<VddModelRef> {
        let mut r = Vec::<VddModelRef>::new();
        let kh = VddKey::EntityModel(0);
        load_by_key_head(&kh).iter().for_each(|x| {
            let model = serde_json::from_slice::<VddModel>(x).unwrap();
            // update_label_cache(model.id, model.name.clone());
            if model.allow_visit && &model.uid != exclude_user {
                r.push(Arc::new(model));
            }
        });
        r
    }

    pub(super) fn m_save_tags(
        batch: &mut WriteBatchWithTransaction<true>,
        model: &VddModel,
        model_tag: &str,
        tag_map: &HashMap<Astr, u64>,
    ) {
        let _ = m_clear_tags(batch, model.id);
        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
        batch.put_cf(
            cf_u,
            VddKey::UniqueModelTag(model_tag.into()).to_string(),
            model.id.to_le_bytes(),
        );
        tag_map.iter().for_each(|x| {
            let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
            batch.put_cf(
                cf_u,
                VddKey::UniqueFieldTag(model_tag.into(), x.0.clone()).to_string(),
                x.1.to_le_bytes(),
            );
        });
    }

    pub(super) fn m_select_tags_to_map(
        with_model: bool,
        with_field: bool,
        model_id: Option<u64>,
    ) -> HashMap<u64, Astr> {
        let key_prefix = VddKey::UniqueModelTag("".into());
        let mut tags = HashMap::<u64, Astr>::new();
        let mut accept = false;
        load_by_key_head_with_key(&key_prefix)
            .iter()
            .for_each(|model_tag_tuple| {
                let mid = u64::from_le_bytes(model_tag_tuple.1.to_vec().try_into().unwrap());
                let model_tag = extract_last_str_from_key(&model_tag_tuple.0);
                accept = model_id.is_none() || (model_id.is_some() && mid == model_id.unwrap());
                if accept && with_model {
                    tags.insert(mid, model_tag.to_owned());
                }
                if accept && with_field {
                    let key_head = VddKey::UniqueFieldTag(model_tag, "".into());
                    load_by_key_head_with_key(&key_head)
                        .iter()
                        .for_each(|field_tag_tuple| {
                            let id =
                                u64::from_le_bytes(field_tag_tuple.1.to_vec().try_into().unwrap());
                            let code = extract_last_str_from_key(&field_tag_tuple.0);
                            let code = code.split(":").last().map(|x| Arc::from(x));
                            tags.insert(id, code.unwrap());
                        });
                }
            });
        tags
    }

    pub(super) fn m_is_all_field_have_tag(model_id: u64, model_tag: &str) -> bool {
        let key_prefix = VddKey::UniqueModelTag(model_tag.into());
        let field_count = field_db::impl_rdb::f_select_field_ids(model_id).len();
        let tag_count = count_in_key_head(&key_prefix, |_| true);
        tag_count == field_count
    }

    pub(super) fn m_clear_tags(batch: &mut WriteBatchWithTransaction<true>, model_id: u64) {
        // 先获取模型对应的旧code
        let key_prefix = VddKey::UniqueModelTag("".into());
        let mut remove_key_vec: Vec<String> = Vec::new();
        load_by_key_head_with_key(&key_prefix)
            .iter()
            .for_each(|model_tag_tuple| {
                let id = u64::from_le_bytes(model_tag_tuple.1.to_vec().try_into().unwrap());
                if id == model_id {
                    remove_key_vec.push(model_tag_tuple.0.to_string());
                    let old_model_code = extract_last_str_from_key(&model_tag_tuple.0);
                    if !old_model_code.is_empty() {
                        //没有找到旧的code说明还没有设置，后续操作可以省略
                        //再获取以旧code开头的字段code
                        let key_prefix = VddKey::UniqueFieldTag(old_model_code, "".into());
                        load_by_key_head_with_key(&key_prefix)
                            .iter()
                            .for_each(|field_tag_tuple| {
                                remove_key_vec.push(field_tag_tuple.0.to_string());
                            });
                        //批量删除这些key
                        remove_key_vec.iter().for_each(|x| {
                            batch.delete(x);
                        });
                    };
                }
            });
    }

    pub(super) fn m_load_by_tag(tag: &str) -> Option<VddModelRef> {
        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
        ROCKSDB
            .get_cf(cf_u, VddKey::UniqueModelTag(tag.into()).to_string())
            .unwrap()
            .map(|v| {
                let id = u64::from_le_bytes(v.to_vec().try_into().unwrap());
                m_load(id)
            })
    }

    pub(super) fn m_is_tag_unique(tag: &str, model_id: u64) -> bool {
        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
        let key = VddKey::UniqueModelTag(tag.into()).to_string();
        match ROCKSDB.get_cf(cf_u, &key).expect(i18n!(E_ROCKSDB).as_ref()) {
            None => true,
            Some(v) => {
                let id = u64::from_le_bytes(v.to_vec().try_into().unwrap());
                id == model_id
            }
        }
    }

    /// 获取回收站中用户相关的所有数据
    pub(super) fn tree_of_trash(uid: UserId, schemas: &[Arc<VddSchema>]) -> Vec<WebTreeNode> {
        let mut r = Vec::<WebTreeNode>::new();
        // 不在回收站的模型和其字段
        schemas.iter().for_each(|s| {
            let mut tree_node = WebTreeNode::from_model_with_code(&s.model, "");
            s.fields.iter().for_each(|f| {
                tree_node.add_child(f.id.0, &f.label, "", Some(json!(f).to_string().into()));
            });
            r.push(tree_node);
        });
        let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
        let mut iter = ROCKSDB.raw_iterator_cf(&cf_trash);
        let mut fv_count_map = HashMap::<u64, u64>::new();
        // 回收站中的模型
        iter.seek_to_first();
        while iter.valid() {
            let (k, v) = iter.item().unwrap();
            let key = String::from_utf8(k.to_vec()).unwrap();
            let key = VddKey::try_from(key.as_str());
            if let Ok(key) = key {
                match key {
                    VddKey::EntityModel(_) => {
                        let model = serde_json::from_slice::<VddModel>(v).unwrap();
                        if model.uid == uid {
                            r.push(WebTreeNode::from_model_with_code(&model, "可回收"));
                        }
                    }
                    VddKey::BoundFieldValue(f, ..) => {
                        let mut count = *fv_count_map.get(&f.0).unwrap_or(&0u64);
                        count += 1;
                        fv_count_map.insert(f.0, count);
                    }
                    _ => {}
                }
            }
            iter.next();
        }
        // 回收站中的字段
        iter.seek_to_first();
        while iter.valid() {
            let (k, v) = iter.item().unwrap();
            let key = String::from_utf8(k.to_vec()).unwrap();
            let key = VddKey::try_from(key.as_str());
            if let Ok(key) = key {
                match key {
                    VddKey::EntityField(_) => {
                        let field = serde_json::from_slice::<VddField>(v).unwrap();
                        r.iter_mut()
                            .find(|node| node.key == field.model_id)
                            .map(|node| {
                                node.add_child(
                                    field.id.0,
                                    &field.label,
                                    &format!("{}", fv_count_map.get(&field.id.0).unwrap_or(&0u64)),
                                    Some(json!(field).to_string().into()),
                                );
                            });
                    }
                    _ => {}
                }
            }
            iter.next();
        }
        r
    }

    /// 从回收站中彻底删除一个模型，包含字段、字段值、值基本信息
    pub(super) fn drop_from_trash(model_id: u64) -> Ares {
        let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
        let mut batch = WriteBatchWithTransaction::<true>::default();
        // 回收站中的本体
        let key = VddKey::EntityModel(model_id).to_string();
        batch.delete_cf(&cf_trash, &key);
        let mut iter = ROCKSDB.raw_iterator_cf(&cf_trash);
        // 回收站中的字段 字段值 值基本信息
        iter.seek_to_first();
        while iter.valid() {
            let (k, v) = iter.item().unwrap();
            let key = String::from_utf8(k.to_vec()).unwrap();
            let key = VddKey::try_from(key.as_str());
            if let Ok(key) = key {
                match key {
                    VddKey::EntityField(_) => {
                        let field = serde_json::from_slice::<VddField>(v).unwrap();
                        if field.model_id == model_id {
                            let key = VddKey::EntityField(field.id.clone()).to_string();
                            batch.delete_cf(&cf_trash, &key);
                            let key_head = VddKey::BoundFieldValue(field.id, EMPTY_RID);
                            delete_by_key_head_in_trash(&mut batch, &key_head);
                        }
                    }
                    VddKey::EntityRecordMeta(_) => {
                        let meta = serde_json::from_slice::<RecordMeta>(v).unwrap();
                        if meta.model_id == model_id {
                            let key = VddKey::EntityRecordMeta(meta.id.clone()).to_string();
                            batch.delete_cf(&cf_trash, &key);
                        }
                    }
                    _ => {}
                }
            }
            iter.next();
        }
        ROCKSDB.write(batch).map_err(|e| e.to_string().into())
    }

    /// 从回收站中恢复模型
    pub(super) fn restore(model_id: u64, field_ids: &[FieldId]) -> Ares {
        let mut batch = WriteBatchWithTransaction::<true>::default();
        let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
        let mut index_changes = Vec::<IndexChange>::new();
        let key = VddKey::EntityModel(model_id).to_string();

        let get_r = ROCKSDB.get_cf(&cf_trash, &key).unwrap(); // 看模型是否在回收站中，如果在，移出回收站
        match get_r {
            Some(buffer) => {
                let m = serde_json::from_slice::<VddModel>(&buffer).unwrap();
                m_save(&mut batch, &mut index_changes, &m);
                batch.delete_cf(&cf_trash, &key);
            }
            _ => {
                tracing::warn!("Restoring model with id {} not found", model_id);
            }
        };

        for field_id in field_ids {
            let key = VddKey::EntityField(field_id.clone()).to_string();
            let field = ROCKSDB.get_cf(&cf_trash, &key).unwrap();
            match field {
                Some(buffer) => {
                    let field = serde_json::from_slice::<VddField>(&buffer).unwrap();
                    field_db::impl_rdb::f_save(&mut batch, &mut index_changes, &field);
                    batch.delete_cf(&cf_trash, &key);
                }
                _ => {
                    tracing::warn!("Restoring field with id {} not found", field_id.0);
                }
            }
        }

        let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
        if ares.is_ok() {
            cache::apply_index_change(&index_changes)
        } else {
            ares
        }
    }

    pub(super) fn design_in_trash(model_id: u64) -> bool {
        let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
        let key = VddKey::EntityModel(model_id).to_string();
        let mut design_in_trash = ROCKSDB.key_may_exist_cf(&cf_trash, &key);
        if !design_in_trash {
            let field_ids = field_db::impl_rdb::f_select_field_ids(model_id);
            for field_id in field_ids.iter() {
                let key = VddKey::EntityField(field_id.clone()).to_string();
                design_in_trash = ROCKSDB.key_may_exist_cf(&cf_trash, &key);
                if design_in_trash {
                    break;
                }
            }
        }
        design_in_trash
    }
}

mod impl_sqlx {
    use crate::field_db;
    use crate::keys::VddKey;
    use crate::sqlx::SqlxDefineTrait;
    use serde_json::json;
    use std::collections::HashMap;
    use std::sync::Arc;
    use visit_dd_core::field::VddField;
    use visit_dd_core::model::{VddModel, VddModelRef, VddSchema};
    use visit_dd_core::query::WebTreeNode;
    use visit_dd_core::{Ares, Astr, FieldId, UserId, EMPTY_FID, EMPTY_UID};

    pub(super) async fn m_load(pool: &impl SqlxDefineTrait, mid: u64) -> VddModelRef {
        let sql_r = pool.load_entity(mid).await.unwrap();
        let r = serde_json::from_str::<VddModel>(&sql_r).unwrap();
        Arc::new(r)
    }

    pub(super) async fn m_select_by_uid(
        pool: &impl SqlxDefineTrait,
        uid: &UserId,
    ) -> Vec<VddModelRef> {
        let sql_r = pool
            .load_entities_form_index_query(VddKey::IndexUser2Model(uid.clone()), false)
            .await
            .unwrap();
        sql_r
            .iter()
            .map(|x| {
                let m = serde_json::from_str::<VddModel>(x).unwrap();
                Arc::new(m)
            })
            .collect::<Vec<VddModelRef>>()
    }

    pub(super) async fn m_select_ids_by_uid(pool: &impl SqlxDefineTrait, uid: &UserId) -> Vec<u64> {
        let ids = pool.read_index(VddKey::IndexUser2Model(uid.clone())).await;
        ids.iter().map(|x| *x as u64).collect::<Vec<u64>>()
    }

    pub(super) async fn m_save(
        pool: &impl SqlxDefineTrait,
        model: &VddModel,
        is_update: bool,
    ) -> Ares {
        if is_update {
            pool.update_entity(model.id, &serde_json::to_string(model).unwrap())
                .await
        } else {
            pool.m_save_new(model).await
        }
    }
    pub(super) async fn m_delete(pool: &impl SqlxDefineTrait, model: &VddModel) -> Ares {
        //删除用户模型索引中的关系
        pool.delete_index(VddKey::IndexUser2Model(model.uid.clone()), model.id)
            .await?;
        // 访问请求
        pool.delete_bounds(VddKey::BoundVisitRequest(model.id, EMPTY_UID))
            .await?;
        // 分享规则
        pool.delete_bounds(VddKey::BoundVisitGroup(model.id, EMPTY_UID))
            .await?;
        // 订阅
        pool.delete_bounds(VddKey::BoundModelAndSubscribe(model.id, EMPTY_UID))
            .await?;
        // 记录规则
        pool.delete_bounds(VddKey::BoundRecordRule(model.id, 0))
            .await?;
        // 标记
        pool.delete_uniques_from_index_query(
            VddKey::UniqueFieldTag("".into(), "".into()),
            VddKey::IndexModel2Field(model.id),
        )
        .await?;
        pool.delete_unique_by_value(VddKey::UniqueModelTag("".into()), model.id)
            .await?;
        //删除字段定义
        pool.delete_entities_from_index_query(VddKey::IndexModel2Field(model.id))
            .await?;
        //删除该模型到字段的索引
        pool.delete_indexes(VddKey::IndexModel2Field(model.id))
            .await?;
        // 删除模型
        pool.delete_entity(model.id).await
    }

    pub(super) async fn m_count_fields(pool: &impl SqlxDefineTrait, model_id: u64) -> usize {
        pool.count_in_index(VddKey::IndexModel2Field(model_id))
            .await
            .unwrap()
    }

    pub(super) async fn m_count_all(pool: &impl SqlxDefineTrait) -> usize {
        pool.count_entities(VddKey::EntityModel(0)).await.unwrap()
    }

    pub(super) async fn m_schema(
        pool: &impl SqlxDefineTrait,
        mid: u64,
    ) -> Result<Arc<VddSchema>, Astr> {
        let model = m_load(pool, mid).await;
        let fields = field_db::f_select_by_model(mid).await;
        let tags = m_select_tags_to_map(pool, true, true, Some(mid)).await;
        let schema = VddSchema {
            model,
            fields,
            tags,
        };
        Ok(Arc::new(schema))
    }

    pub(super) async fn m_select_sharing(
        pool: &impl SqlxDefineTrait,
        exclude_user: &UserId,
    ) -> Vec<VddModelRef> {
        let sql_r = pool
            .load_entities_form_index_query(VddKey::IndexUser2Model(UserId(0)), false)
            .await
            .unwrap();
        sql_r
            .iter()
            .map(|x| Arc::new(serde_json::from_str::<VddModel>(x).unwrap()))
            .filter(|x| x.allow_visit && &x.uid != exclude_user)
            .collect::<Vec<VddModelRef>>()
    }

    pub(super) async fn m_select_tags_to_map(
        pool: &impl SqlxDefineTrait,
        with_model: bool,
        with_field: bool,
        model_id: Option<u64>,
    ) -> HashMap<u64, Astr> {
        let mut r = HashMap::<u64, Astr>::new();
        let uniques = pool
            .select_uniques_by_value(VddKey::UniqueModelTag("".into()), model_id)
            .await;
        if with_model {
            uniques.iter().for_each(|x| {
                r.insert(*x.1, x.0.clone().into());
            })
        }
        if with_field {
            for x in uniques.iter() {
                let sub_uniques = pool
                    .select_uniques_from_index_query(
                        VddKey::UniqueFieldTag("".into(), "".into()),
                        VddKey::IndexModel2Field(*x.1),
                    )
                    .await;
                sub_uniques.iter().for_each(|y| {
                    r.insert(
                        *y.1,
                        y.0.clone().split(":").last().unwrap().to_string().into(),
                    );
                })
            }
        }
        r
    }
    pub(super) async fn m_is_all_field_have_tag(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
        _tag: &str,
    ) -> bool {
        let model = m_load(pool, model_id).await;
        let field_ids = field_db::f_select_field_ids(model.id)
            .await
            .iter()
            .map(|x| x.0 as i64)
            .collect::<Vec<i64>>();
        let uniques = pool
            .select_uniques_from_index_query(
                VddKey::UniqueFieldTag("".into(), "".into()),
                VddKey::IndexModel2Field(model_id),
            )
            .await;
        uniques.len() == field_ids.len()
    }

    pub(super) async fn m_load_by_tag(
        pool: &impl SqlxDefineTrait,
        tag: &str,
    ) -> Option<VddModelRef> {
        let uniques = pool
            .select_uniques(VddKey::UniqueModelTag(tag.into()))
            .await;
        if uniques.is_empty() {
            return None;
        }
        let mid = uniques.get(tag).unwrap();
        let model = m_load(pool, *mid).await;
        Some(model)
    }
    pub(super) async fn m_is_tag_unique(
        pool: &impl SqlxDefineTrait,
        tag: &str,
        model_id: u64,
    ) -> bool {
        let uniques = pool
            .select_uniques_by_value(VddKey::UniqueModelTag("".into()), Some(model_id))
            .await;
        let mid = uniques.get(tag);
        mid.is_none() || mid.unwrap() == &model_id
    }

    pub(super) async fn tree_of_trash(
        pool: &impl SqlxDefineTrait,
        uid: UserId,
        schemas: &[Arc<VddSchema>],
    ) -> Vec<WebTreeNode> {
        let mut r = Vec::<WebTreeNode>::new();
        // 不在回收站的模型和其字段
        schemas.iter().for_each(|s| {
            let mut tree_node = WebTreeNode::from_model_with_code(&s.model, "");
            s.fields.iter().for_each(|f| {
                tree_node.add_child(f.id.0, &f.label, "", Some(json!(f).to_string().into()));
            });
            r.push(tree_node);
        });
        pool.load_entities(VddKey::EntityModel(0), true)
            .await
            .map(|jsons| {
                jsons.iter().for_each(|json| {
                    let model = serde_json::from_str::<VddModel>(json).unwrap();
                    if model.uid == uid {
                        r.push(WebTreeNode::from_model_with_code(&model, "可回收"));
                    }
                });
            })
            .unwrap();
        pool.load_entities(VddKey::EntityField(EMPTY_FID), true)
            .await
            .map(|jsons| {
                jsons.iter().for_each(|json| {
                    let field = serde_json::from_str::<VddField>(json).unwrap();
                    r.iter_mut()
                        .find(|node| node.key == field.model_id)
                        .map(|node| {
                            node.add_child(field.id.0, &field.label, "可回收", Some(json.clone()));
                        });
                })
            })
            .unwrap();
        r
    }

    /// 删除回收站中的模型，返回删除的字段id
    pub(super) async fn m_drop_from_trash(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
    ) -> Result<Vec<FieldId>, Astr> {
        let mut field_ids = Vec::<FieldId>::new();
        for field_j in pool
            .load_entities(VddKey::EntityField(EMPTY_FID), true)
            .await
            .unwrap()
            .iter()
        {
            let field = serde_json::from_str::<VddField>(field_j).unwrap();
            if field.model_id == model_id {
                field_ids.push(field.id.clone());
                pool.do_delete_entity(field.id.0).await?;
            }
        }
        pool.do_delete_entity(model_id).await?;
        Ok(field_ids)
    }

    pub(super) async fn m_restore(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
        field_ids: &[FieldId],
        operator: UserId,
    ) -> Ares {
        pool.restore_entity(model_id).await?;
        pool.add_to_index(VddKey::IndexUser2Model(operator), model_id)
            .await?;
        for field_id in field_ids.iter() {
            pool.restore_entity(field_id.0).await?;
            pool.add_to_index(VddKey::IndexModel2Field(model_id), field_id.0)
                .await?;
        }
        Ok(())
    }

    pub(super) async fn design_in_trash(pool: &impl SqlxDefineTrait, model_id: u64) -> bool {
        let mut b = pool.check_entity_deleted(model_id).await;
        if !b {
            let field_ids = field_db::f_select_field_ids(model_id).await;
            for field_id in field_ids.iter() {
                b = pool.check_entity_deleted(field_id.0).await;
                if b {
                    break;
                }
            }
        }
        b
    }
}
